package xyz.codej.websocket.netty.server;

import cn.hutool.core.net.NetUtil;
import cn.hutool.core.thread.ThreadUtil;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.util.concurrent.ImmediateEventExecutor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import xyz.codej.websocket.netty.server.config.NettyServerConfig;
import xyz.codej.websocket.utils.Server;
import xyz.codej.websocket.utils.ServerRunnable;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @ClassName: NettyServer
 * @Description Netty 服务启动监听器
 * @Author 王华君
 * @Date 2020/4/1 5:02 下午
 * @VERSION 1.0
 */
@Component
@Slf4j
public class NettyServer implements Server {
    @Autowired
    NettyServerConfig serverConfig;

    private EventLoopGroup bossGroup = new NioEventLoopGroup(1);
    private EventLoopGroup workerGroup = new NioEventLoopGroup();

    /**
     * 使用DefaultChannelGroup保存所有WebSocket客户端
     * ChannelGroup不需要remove,系统会自动管理
     */
    private final ChannelGroup channelGroup = new DefaultChannelGroup(ImmediateEventExecutor.INSTANCE);
    private final ConcurrentHashMap<String, ChannelId> clientChannelMap = new ConcurrentHashMap<>();

    /**
     * 创建Channel初始化器
     */
    @Autowired
    WebSocketServerInitializer webSocketServerInitializer;

    /**
     * 使用线程启动netty
     */
    @PostConstruct
    public void asyncStart() {
        ThreadUtil.execAsync(new ServerRunnable(this));
    }

    /**
     * 同步启动 Netty
     *
     * @return
     * @throws InterruptedException
     */
    @Override
    public void start() throws InterruptedException {
        /***
         * NioEventLoopGroup 是用来处理I/O操作的多线程事件循环器，
         * Netty提供了许多不同的EventLoopGroup的实现用来处理不同传输协议。 在这个例子中我们实现了一个服务端的应用，
         * 因此会有2个NioEventLoopGroup会被使用。 第一个经常被叫做‘boss’，用来接收进来的连接。
         * 第二个经常被叫做‘worker’，用来处理已经被接收的连接， 一旦‘boss’接收到连接，就会把连接信息注册到‘worker’上。
         * 如何知道多少个线程已经被使用，如何映射到已经创建的Channels上都需要依赖于EventLoopGroup的实现，
         * 并且可以通过构造函数来配置他们的关系。
         */

        /**
         * ServerBootstrap 是一个启动NIO服务的辅助启动类 你可以在这个服务中直接使用Channel
         */
        ServerBootstrap b = new ServerBootstrap()

                /**
                 * 这一步是必须的，如果没有设置group将会报java.lang.IllegalStateException: group not
                 * set异常
                 */
                .group(bossGroup, workerGroup)

                /***
                 * ServerSocketChannel以NIO的selector为基础进行实现的，用来接收新的连接
                 * 这里告诉Channel如何获取新的连接.
                 */
                .channel(NioServerSocketChannel.class)

                /***
                 * childHandler会在客户端成功connect后才执行
                 *
                 * 这里的事件处理类经常会被用来处理一个最近的已经接收的Channel。 ChannelInitializer是一个特殊的处理类，
                 * 他的目的是帮助使用者配置一个新的Channel。
                 * 也许你想通过增加一些处理类比如NettyServerHandler来配置一个新的Channel
                 * 或者其对应的ChannelPipeline来实现你的网络程序。 当你的程序变的复杂时，可能你会增加更多的处理类到pipline上，
                 * 然后提取这些匿名类到最顶层的类上。
                 */
                .childHandler(webSocketServerInitializer
                        .setChannelGroup(channelGroup)
                        .setClientChannelMap(clientChannelMap))

                /**
                 * 设置队列大小
                 */
                .option(ChannelOption.SO_BACKLOG, 1024)

                /**
                 * 两小时内没有数据的通信时,TCP会自动发送一个活动探测数据报文
                 */
                .childOption(ChannelOption.SO_KEEPALIVE, true)

                /**
                 * handler在初始化时就会执行
                 */
                .handler(new LoggingHandler(LogLevel.INFO));

        /***
         * 绑定ip、端口并启动去接收进来的连接
         */
        ChannelFuture future = b.bind(NetUtil.createAddress(serverConfig.getAddress(), serverConfig.getPort()));
        log.info("启动netty");
        /**
         * 这里会一直等待，直到socket被关闭
         */
        future.channel().closeFuture().sync();
    }

    /**
     * 销毁
     */
    @PreDestroy
    public void destroy() {
        log.info("关闭netty");
        channelGroup.close();
        bossGroup.shutdownGracefully();
        workerGroup.shutdownGracefully();
    }

    public void boardCast(String message) {
        channelGroup.writeAndFlush(message);
    }

    public void sendMessage(String reqId, String message) {
        ChannelId channelId = clientChannelMap.get(reqId);
        if (channelId != null) {
            Channel channel = channelGroup.find(channelId);
            if(channel!=null) {
                ChannelFuture cf = channel.writeAndFlush(new TextWebSocketFrame(message));
                cf.addListener(new ChannelFutureListener() {
                    @Override
                    public void operationComplete(ChannelFuture future) {
                        if (!future.isSuccess()) {
                            log.info("发送失败");
                        }else{
                            log.info("发送成功");
                            removeClientChannelMap(channelId);
                        }
                    }
                });
            }
        }else{
            log.error("客户端未找到");
        }
    }

    /**
     * 移除 reqId 和 channel 的对应关系
     * @param channelId
     */
    public void removeClientChannelMap(ChannelId channelId){
        for (Map.Entry<String, ChannelId> entry : clientChannelMap.entrySet()) {
            if (entry.getValue().equals(channelId)) {
                clientChannelMap.remove(entry.getKey());
                break;
            }
        }
    }
}
